//package com.credithc.signature.job.canal;
//
//import com.alibaba.fastjson.JSON;
//import com.credithc.signature.common.po.SignatureLogExt;
//import com.credithc.signature.common.ro.core.CoreRequet;
//import com.credithc.signature.common.service.SignatureLogExtService;
//import com.credithc.signature.job.constant.CanalConstant;
//import com.credithc.signature.job.constant.MQConstant;
//import com.credithc.signature.job.mq.JobMQSender;
//import com.credithc.signature.job.po.CanalPo;
//import com.credithc.signature.job.ro.JobRequest;
//import com.credithc.signature.job.service.SignatureService;
//import com.credithc.signature.job.utils.ProtobufUtil;
//import com.credithc.signature.job.utils.SpringUtil;
//import com.rabbitmq.client.Channel;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.collections.CollectionUtils;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.core.MessageBuilder;
//import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
//import org.springframework.beans.BeanUtils;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.util.ArrayList;
//import java.util.List;
//
///**
// * @desc: canal消费者类<br />
// * @author: Jingbo Hu<br/>
// * @time：2021年07⽉28⽇<br/>
// * @version：v1.0.0<br>
// */
//@Slf4j
////@Component
//public class CanalMqConsumer implements ChannelAwareMessageListener {
//
//    @Autowired
//    SignatureService signatureService;
//
//    @Autowired
//    SignatureLogExtService logExtService;
//
//    @Autowired
//    JobMQSender mqSender;
//
//
//    @Override
//    public void onMessage(Message message, Channel channel) throws Exception {
//
//        try {
//            byte[] body = message.getBody();
//            String jsonBody = new String(body);
//            CanalPo canalPo = JSON.parseObject(jsonBody, CanalPo.class);
//            String database = canalPo.getDatabase();
//            String type = canalPo.getType();
//            String tableName = canalPo.getTable();
//            // 解析并判断是否是 signature->request_log->update操作；如果否：确认即可，如果是：判断签章状态：失败的话进行重签操作
//            if (CanalConstant.DATABASE.equalsIgnoreCase(database) && CanalConstant.SQLTYPE.equalsIgnoreCase(type) && tableName.contains(CanalConstant.TABLENAMEPREFIX)) {
//                // 过滤不需要重签的数据
//                List<CanalPo.DataDTO> data = filterData(canalPo.getData());
//                if (CollectionUtils.isNotEmpty(data)) {
//                    for (CanalPo.DataDTO dataDTO : data) {
//                        log.info("=====================Receive Canal RabbitMq message=====================");
//                        CoreRequet coreRequet = new CoreRequet();
//                        BeanUtils.copyProperties(dataDTO, coreRequet);
//                        SignatureLogExt logExt = logExtService.selectOneById(dataDTO.getRefId(), dataDTO.getCreateTime());
//                        coreRequet.setJsonBody(logExt.getRequestData());
//                        coreRequet.setCallbackUrl(dataDTO.getCallbackUrl());
//                        coreRequet.setNeedSign(dataDTO.getIsSeal().equals("1") ? true : false);
//                        byte[] objectBytes = ProtobufUtil.serializer(coreRequet);
//                        mqSender.sendSync(MessageBuilder.withBody(objectBytes).build());
//                    }
//                }
//            }
//        } catch (Throwable e) {
//            log.error("canal消息处理失败重新发回MQ, {}\n", e.getMessage());
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//        } finally {
//            try {
//                // 确认消息成功接收
//                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//            } catch (Throwable e) {
//                // 不做处理
//                log.error(e.getMessage());
//            }
//        }
//    }
//
//    /**
//     * 过滤不需要重签的数据
//     *
//     * @param data
//     * @return
//     */
//    private List<CanalPo.DataDTO> filterData(List<CanalPo.DataDTO> data) {
//        List<CanalPo.DataDTO> newData = new ArrayList<>();
//        data.stream().forEach(t -> {
//            String signStatus = t.getSignStatus();
//            if (CanalConstant.FAILED.equals(signStatus)) {
//                String signResult = t.getSignResult();
//                if (signResult.contains(CanalConstant.FREEMARKER_ERROR) ||
//                        signResult.contains(CanalConstant.IDNO_ERROR) ||
//                        signResult.contains(CanalConstant.SEALNAMEEMPTY_ERROR) ||
//                        signResult.contains(CanalConstant.SEALCODE_ERROR)) {
//                    String requestSn = t.getRequestSn();
//                    log.info("requestNo:{}, 此数据不能重签，理由：{}", requestSn, signResult);
//                } else {
//                    newData.add(t);
//                }
//            }
//        });
//        return newData;
//    }
//
//}
